package com.assistant.test;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.Serializable;

public class Example {

    public static void main(String[] args) throws Exception {
        //1.创建flink的执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//2.创建数据源
        DataStream<Person> flintstones = env.fromElements(
            new Person("Fred", 35),
            new Person("Wilma", 35),
            new Person("Pebbles", 2));
		//3.对数据进行过滤
        DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
            @Override
            public boolean filter(Person person) throws Exception {
                return person.age >= 18;
            }
        });
		//4.数据的处理方式（输出控制台）
        adults.print();
		//5.执行flink任务
        env.execute();
    }

    public static class Person implements Serializable {
        public String name;
        public Integer age;
        public Person() {}
        public Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }
        public String toString() {
            return this.name.toString() + ": age " + this.age.toString();
        }
    }

}